#include "TCPClient.h"
#include "Connector.h"
#include "Socket.h"
#include "sock.h"
#include "TCPConnection.h"
#include "eventloop/Eventloop.h"
#include "log/Logger.h"
#include "utility/assertion.h"

#include <array>

TCPClient::TCPClient(Eventloop &loop, const Sockaddr &addr, std::string name, bool retry):
    _mutex{},
    _name{std::move(name)},
    _connectionCallback{Callback::default_connection},
    _drainCallback{default_connection_func},
    _messageCallback{Callback::default_message},
    _connector{std::make_shared<Connector>(loop, addr)},
    _connection{},
    _loop{loop},
    _connectionID{0},
    _retry{retry},
    _connecting{true} {
    _connector->set_new_connection_callback(
        [this](std::unique_ptr<Socket> &&socket) {
            _new_connection_callback(std::move(socket));
    });
    LOG_INFO << "TCPClient [" << _name << "]:connector " << _connector.get()
             << " constructed";
}

TCPClient::~TCPClient() {
    LOG_INFO << "TCPClient [" << _name << "]:connector " << _connector.get()
             << " destroying";
    long count = 0;
    TCPConnectionPtr connection{};
    {
        const std::lock_guard guard{_mutex};
        count = _connection.use_count();
        connection = _connection;
    }
    _destroy_connection(connection, count);
    LOG_DEBUG << "TCPClient [" << _name << "]:connector " << _connector.get()
              << " destroyed " << count;
}

void TCPClient::connect() {
    LOG_INFO << "TCPClient [" << _name << "]:connecting to "
             << _connector->addr().IP_port();
    _connecting = true;
    _connector->start();
}

void TCPClient::disconnect() {
    _connecting = false;
    const std::lock_guard guard{_mutex};
    if (!_connection) {
        return;
    }
    _connection->shutdown();
}

void TCPClient::send(const std::string &message) {
    const std::lock_guard guard{_mutex};
    if (!_connection) {
        return;
    }
    _connection->send(message);
}

void TCPClient::send(std::string &&message) {
    const std::lock_guard guard{_mutex};
    if (!_connection) {
        return;
    }
    _connection->send(std::move(message));
}

void TCPClient::stop() {
    _connecting = false;
    _connector->stop();
}


void TCPClient::_close_callback(const TCPConnectionPtr &connection) {
    _loop.ASSERT_IN_THIS_THREAD();
    assert(_loop == connection->loop());
    {
        const std::lock_guard guard{_mutex};
        assert(_connection == connection);
        _connection.reset();
    }
    _loop.queue_in_loop([connection](){ connection->destroy(); });
    if (!_retry || !_connecting) {
        return;
    }
    LOG_INFO << "TCPClient::connect[" << _name << "]:reconnecting to "
             << _connector->addr().IP_port();
    _connector->restart();
}

void TCPClient::_destroy_connection(const TCPConnectionPtr &connection, long count) {
    if (!connection) {
        _connector->stop();
        _loop.queue_in_loop([connector = _connector](){});
        return;
    }
    assert(_loop == connection->loop());
    _loop.run_in_loop([connection](){
        connection->set_close_callback([](const TCPConnectionPtr &connectionPtr){
            connectionPtr->destroy();
        });
    });
    if (count == 1) {
        connection->force_close();
    }
}

void TCPClient::_new_connection_callback(std::unique_ptr<Socket> &&socket) {
    _loop.ASSERT_IN_THIS_THREAD();
    std::array<char, 1 << 5> buffer{};
    const auto fd = socket->fd();
    const Sockaddr peer{Sock::peer_addr(fd)};
    ::snprintf(
        buffer.data(), buffer.size(), ":%s#%d", peer.IP_port().c_str(), 
        ++_connectionID);
    const auto connection = std::make_shared<TCPConnection>(
        _loop, _name + buffer.data(), std::move(socket), 
        Sockaddr{Sock::local_addr(fd)}, peer);
    connection->set_connection_callback(_connectionCallback);
    connection->set_drain_callback(_drainCallback);
    connection->set_message_callback(_messageCallback);
    connection->set_close_callback(
        [this](const TCPConnectionPtr &connectionPtr) {
            _close_callback(connectionPtr);
    });
    {
        const std::lock_guard guard{_mutex};
        _connection = connection;
    }
    connection->established();
}
